package com.ls.mq.manager;

import java.util.ArrayList;
import java.util.List;

import com.ls.mq.iterface.MessageListener;
import com.ls.mq.listener.impl.DefaultMessageListener;
import com.ls.mq.pubsub.PubSubProxy;
import com.ls.mq.queue.QueueProxy;
import com.ls.mq.redis.RedisFactory;
import com.sun.istack.internal.Nullable;

/**
 * 负责管理Mq启动，停止，注册和删除监听者 mq包括pubsub,queue
 * <p>
 * 负责初始化RedisFactory，添加channels后启动
 * <p>
 * 对于redis参数，由哪些channel等这些参数的获取，不放到该类里面，放到上层service调用，通过这样来区分功能，结构清晰
 * 
 * @author Administrator
 *
 */
public class MQManager {

	private List<String> queueChannels;
	private List<String> pubsubChannels;

	private volatile static MQManager instance;
	private MessageListener defaultQueueListener;
	private MessageListener defaultPubSubListener;

	public static MQManager initInstance(String hostName, int port, @Nullable String userName,
			@Nullable String password) {
		if (instance == null)
			synchronized (MQManager.class) {
				if (instance == null)
					instance = new MQManager(hostName, port, userName, password);
			}
		return instance;
	}

	public static MQManager getInstance(){
		if(instance==null)
			throw new IllegalStateException("初始化未完成");
		return instance;
	}
	
	public void start() {
		startQueue();
		startPubSub();
	}

	public void stop(){
		QueueProxy.getInstance().stop();
		PubSubProxy.getInstance().stop();
	}
	
	private void startQueue() {
		for (String c : getQueueChannels()) {
			addQueueListener(c, defaultQueueListener);
		}
	}

	private void startPubSub() {
		addPubSubListener(defaultPubSubListener, getPubsubChannels().toArray(new String[0]));
	}

	public MQManager(String hostName, int port, String userName, String password) {
		RedisFactory.initInstance(hostName, port, userName, password);
		queueChannels = new ArrayList<>();
		pubsubChannels = new ArrayList<>();
		defaultQueueListener = new DefaultMessageListener();
		defaultPubSubListener = new DefaultMessageListener();
	}

	public void addQueueListener(String channel, MessageListener listener) {
		QueueProxy.getInstance().addListener(channel, listener);
	}

	public void addPubSubListener(MessageListener listener, String... channels) {
		PubSubProxy.getInstance().addListener(listener, channels);
	}

	public void addPubSubListener(String channel, MessageListener listener) {
		PubSubProxy.getInstance().addListener(channel, listener);
	}

	public void removeQueueListener(String channel) {
		QueueProxy.getInstance().removeListener(channel);
	}

	public void removePubSubListener(String channel) {
		PubSubProxy.getInstance().removeListener(channel);
	}

	public List<String> getQueueChannels() {
		return queueChannels;
	}

	public void setQueueChannels(List<String> queueChannels) {
		this.queueChannels = queueChannels;
	}

	public List<String> getPubsubChannels() {
		return pubsubChannels;
	}

	public void setPubsubChannels(List<String> pubsubChannels) {
		this.pubsubChannels = pubsubChannels;
	}

}
